package com.qf.leadnewsschedule.service.impl;


import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.qf.leadnewsschedule.mapper.TaskInfoMapper;
import com.qf.leadnewsschedule.service.TaskinfoLogsService;
import com.qf.leadnewsschedule.service.TaskinfoService;
import com.qf.qfleadnewsmodel.commons.consts.RedisConst;
import com.qf.qfleadnewsmodel.enums.TaskTypeEnum;
import com.qf.qfleadnewsmodel.schedule.dtos.Task;
import com.qf.qfleadnewsmodel.schedule.pojos.Taskinfo;
import com.qf.qfleadnewsmodel.schedule.pojos.TaskinfoLogs;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class TaskinfoServiceImpl extends ServiceImpl<TaskInfoMapper, Taskinfo> implements TaskinfoService {

    @Autowired
    private TaskinfoLogsService taskinfoLogsService;

    @Autowired
    private ApplicationContext ioc;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    @Override
    public void add(Task task) {
        Taskinfo taskinfo = ioc.getBean(TaskinfoService.class).save2DB(task);

        if (taskinfo != null){
            save2Redis(taskinfo);
        }
    }

    private void save2Redis(Taskinfo taskinfo) {
        Calendar instance = Calendar.getInstance();
        //基于当前时间加5分钟
        instance.add(Calendar.MINUTE,5);
        //未来5分钟的毫秒值
        long futrue5MinMillis = instance.getTime().getTime();
        //任务执行时间的毫秒值
        long taskExeMillis = taskinfo.getExecuteTime().getTime();

        if (taskExeMillis <= futrue5MinMillis){
            //添加到redis
            //当前系统时间毫秒值
            long nowMillis = System.currentTimeMillis();
            if (taskExeMillis<=nowMillis){
                //加入即时消费的队列中
                String key = buildKey(taskinfo,RedisConst.LIST_TYPE);

                redisTemplate.opsForList().leftPush(key, JSON.toJSONString(taskinfo));
            }else{
                //加入到预热的队列中
                String key = buildKey(taskinfo, RedisConst.ZSET_TYPE);

                redisTemplate.opsForZSet().add(key,JSON.toJSONString(taskinfo),taskExeMillis);
            }
        }
    }

    /**
     * 根据不同队列类型构建该队列类型的key
     *
     * @param taskinfo
     * @param type
     * @return
     */
    private String buildKey(Taskinfo taskinfo, String type) {
        if (type == RedisConst.LIST_TYPE) {
            return RedisConst.NOW_CONSUME_TASK_PREFIX + taskinfo.getTaskType() + "_" + taskinfo.getPriority();
        }else if (type == RedisConst.ZSET_TYPE){
            return RedisConst.FUTURE_CONSUME_TASK_PREFIX + taskinfo.getTaskType() + "_" + taskinfo.getPriority();
        }

        return null;
    }

    @Override
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public Taskinfo save2DB(Task task) {
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task,taskinfo);
        //执行时间额外处理
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));

        save(taskinfo);

        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo,taskinfoLogs);
        taskinfoLogs.setStatus(0);
        taskinfoLogsService.save(taskinfoLogs);

        return taskinfo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Task consume(Integer taskType, Integer priority) {

        //因为这个操作执行频读比较高，所以这里加读锁
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(RedisConst.DATA_MOVE_AND_CONSUME_READ_WRITE_LOCK);

        RLock lock = readWriteLock.readLock();
        //获取key
        Taskinfo taskinfo = new Taskinfo();
        taskinfo.setTaskType(taskType);
        taskinfo.setPriority(priority);
        String key = buildKey(taskinfo, RedisConst.LIST_TYPE);

        try {
            lock.lock();
            String taskStr = redisTemplate.opsForList().rightPop(key);
            taskinfo = JSON.parseObject(taskStr, Taskinfo.class);
            if (taskinfo != null) {
                //获取到了
                Task task = new Task();
                BeanUtils.copyProperties(taskinfo, task);
                task.setExecuteTime(taskinfo.getExecuteTime().getTime());

                //数据库的数据重新维护
                removeById(taskinfo.getTaskId());

                TaskinfoLogs logs = new TaskinfoLogs();
                logs.setStatus(1);
                logs.setTaskId(taskinfo.getTaskId());
                taskinfoLogsService.updateById(logs);

                return task;
            }
        }finally {
            lock.unlock();
        }

        return null;
    }

    @Scheduled(cron = "0/5 * * * * ?")
    public void zset2List(){
        log.info("----开始zset向list队列迁移---- 时间：{}", LocalDateTime.now());
        //先获取所有zset相关的键值（key）
        TaskTypeEnum[] taskTypes = TaskTypeEnum.values();

        List<String> futrueKeys = Arrays.stream(taskTypes).map(taskTypeEnum -> {
            int type = taskTypeEnum.getTaskType();
            int priority = taskTypeEnum.getPriority();

            Taskinfo taskInfo = new Taskinfo();
            taskInfo.setTaskType(type);
            taskInfo.setPriority(priority);
            String futrueKey = buildKey(taskInfo, RedisConst.ZSET_TYPE);
            return futrueKey;
        }).collect(Collectors.toList());

        if (futrueKeys != null){
            futrueKeys.forEach(futrueKey->{
                //将这个futrueKey中，符合条件的任务迁移到list中(score小于等于当前系统时间)
                Set<String> values = redisTemplate.opsForZSet().rangeByScore(futrueKey, 0, System.currentTimeMillis());
                if (values!=null && values.size()>0) {
                    //将这些值从zset中删除，添加到list中（先加再删可能导致数据重复）
                    redisTemplate.opsForZSet().remove(futrueKey, values.toArray());

                    //将数据添加到list中
                    String nowKey = futrueKey.replaceAll(RedisConst.FUTURE_CONSUME_TASK_PREFIX,RedisConst.NOW_CONSUME_TASK_PREFIX);
                    redisTemplate.opsForList().leftPushAll(nowKey,values.toArray(new String[0]));
                }
            });
        }
    }


    @Scheduled(cron = "0 0/5 * * * ?")
    public void db2Redis(){
        log.info("----开始数据库向redis队列迁移---- 时间：{}", LocalDateTime.now());
        //查询数据库中符合条件的数据（执行时间在未来5分钟内的）
        Calendar instance = Calendar.getInstance();
        instance.add(Calendar.MINUTE,5);

        LambdaQueryWrapper<Taskinfo> qw = Wrappers.lambdaQuery(Taskinfo.class);
        qw.le(Taskinfo::getExecuteTime,instance.getTime());

        //先将之前redis中的数据清除
        TaskTypeEnum[] taskTypes = TaskTypeEnum.values();
        Arrays.stream(taskTypes).forEach(taskTypeEnum -> {
            int type = taskTypeEnum.getTaskType();
            int priority = taskTypeEnum.getPriority();

            Taskinfo taskInfo = new Taskinfo();
            taskInfo.setTaskType(type);
            taskInfo.setPriority(priority);
            String futrueKey = buildKey(taskInfo, RedisConst.ZSET_TYPE);
            String nowKey = buildKey(taskInfo, RedisConst.LIST_TYPE);

            redisTemplate.delete(futrueKey);
            redisTemplate.delete(nowKey);
        });

        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(RedisConst.DATA_MOVE_AND_CONSUME_READ_WRITE_LOCK);
        RLock lock = readWriteLock.writeLock();

        try {
            lock.lock();
            List<Taskinfo> taskinfos = list(qw);

            //再将从数据库中获取最新符合要求的数据添加到redis
            taskinfos.forEach(taskinfo -> {
                save2Redis(taskinfo);
            });
        }finally {
            lock.unlock();
        }
    }

}
